package com.example.hbase.util;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.example.hbase.entity.HbaseBaseFilter;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;


import java.io.IOException;
import java.util.*;


@Slf4j
@Component
public final class HbaseUtil {
    private   Connection connection;
    @Value("${hase_root_dir}")
    private   String hase_root_dir;
    @Value("${zk_host}")
    private   String zk_host;
    @Value("${zk_port}")
    private   String zk_port;
    private   Configuration configuration;

    private HbaseUtil(){
    }




    public   void getInstance() {
        if(configuration==null) {
            try {
                configuration = HBaseConfiguration.create();
                configuration.set("hbase.rootdir", hase_root_dir);
                configuration.set("hbase.zookeeper.quorum", zk_host);
                configuration.set("hbase.zookeeper.property.clientPort", zk_port);
                configuration.set("zookeeper.znode.parent", "/hbase");
                connection = ConnectionFactory.createConnection(configuration);
            } catch (IOException e) {
                log.error("", e);
                throw new RuntimeException("连接hbase异常");
            }
        }
    }
    /**
     * 删除一张表
     * @param tableName
     */
    public   void dropTable(String tableName) {
        try {
            if(connection == null){
                getInstance();
            }
            Admin admin = connection.getAdmin();
            if (!admin.isTableAvailable(TableName.valueOf(tableName))) {
                 return;
            }
            admin.disableTable(TableName.valueOf(tableName) );
            admin.deleteTable(TableName.valueOf(tableName));
        } catch (Exception e) {
            log.error("删除hbase error :{}",e.getMessage());
            e.printStackTrace();
        }
    }
    public   Table getTable(String tableName) {
        try {
            if(connection == null){
                getInstance();
            }
            return connection.getTable(TableName.valueOf(tableName));
        } catch (IOException e) {
            log.error("", e);
        }
        return null;
    }
    public   Object getTables() throws IOException {
        if(connection == null){
            getInstance();
        }
           HTableDescriptor[] hTableDescriptors = connection.getAdmin().listTables();
        Map<String,List<String>> tables=new HashMap<>();
        for (HTableDescriptor hTableDescriptor : hTableDescriptors) {
            final ArrayList<String> familyList = new ArrayList<>();
            final Collection<HColumnDescriptor> families = hTableDescriptor.getFamilies();
            for (HColumnDescriptor family : families) {
                familyList.add(family.getNameAsString());
            }
            tables.put(hTableDescriptor.getTableName().toString(),familyList);
        }
        return tables;
    }
    public   List<Map<String, Object>> scan(  String tableName) throws Exception {
        if(connection == null){
            getInstance();
        }
        //通过连接获得表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        //使用scan方法从表中得到数据
//        //初始化行过滤器，使用二进制比较器，大于等于row10
        Scan scan = new Scan();
        RowFilter rf = new RowFilter(CompareFilter.CompareOp.GREATER_OR_EQUAL,
                new BinaryComparator(Bytes.toBytes("row010")));
        final FirstKeyOnlyFilter firstKeyOnlyFilter = new FirstKeyOnlyFilter();
        scan.setFilter(firstKeyOnlyFilter);
//        scan.setStopRow(Bytes.toBytes("10"));

        ResultScanner rs = table.getScanner(scan);
        //打印显示结果行的所有细胞
        List<Map<String, Object>> list = new ArrayList<>();
        getResult(rs, list);
        return list;
    }

    public   Object filterForFamily(String tableName, String columnName) throws IOException {
        if(connection == null){
            getInstance();
        }
        //通过连接获得表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        FamilyFilter ff = new FamilyFilter(CompareFilter.CompareOp.EQUAL ,
                new BinaryComparator(Bytes.toBytes(columnName)));
        Scan scan = new Scan();
        // 通过scan.addFamily(family)  也可以实现此操作
        scan.setFilter(ff);
        ResultScanner resultScanner = table.getScanner(scan);
        List<Map<String, Object>> list = new ArrayList<>();
        getResult(resultScanner, list);
        return list;
    }
    public   Object valueFilter(String tableName, String value) throws IOException {
        if(connection == null){
            getInstance();
        }
        //通过连接获得表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        Filter filter1 = new ValueFilter(CompareFilter.CompareOp.EQUAL,
                new SubstringComparator(value));

        Scan scan = new Scan();
        // 通过scan.addFamily(family)  也可以实现此操作
        scan.setFilter(filter1);
        ResultScanner resultScanner = table.getScanner(scan);
        List<Map<String, Object>> list = new ArrayList<>();
        getResult(resultScanner, list);
        return list;
    }
    public   Object scanQualifiler(String tableName, String columnName) throws IOException {
        if(connection == null){
            getInstance();
        }
        //通过连接获得表对象
        Table table = connection.getTable(TableName.valueOf(tableName));
        QualifierFilter ff = new QualifierFilter(
                CompareFilter.CompareOp.EQUAL , new BinaryComparator(Bytes.toBytes(columnName)));
        Scan scan = new Scan();
        // 通过scan.addFamily(family)  也可以实现此操作
        scan.setFilter(ff);
        ResultScanner resultScanner = table.getScanner(scan);
        List<Map<String, Object>> list = new ArrayList<>();
        getResult(resultScanner, list);
        return list;
    }
    public   List<Map<String, Object>>  scan(String tableName, JSONArray jsonArray, boolean reversed, String startIndex, String endIndex) throws Exception {
        if(connection == null){
            getInstance();
        }
        //通过连接获得表对象
        final TableName tableName1 = TableName.valueOf(tableName);
        Table table = connection.getTable(tableName1);
        //使用scan方法从表中得到数据
        Scan scan = new Scan();
//        //初始化行过滤器，使用二进制比较器，大于等于row10

        // [{"memberId":{"memberId": "1123729"}},"mobile":[{"mobile":"13691257842"},{"mobile2":"sss"}],{"memberId":{"535114"}}]
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        jsonArray.forEach(o->{
            JSONObject jsonObject = JSONObject.parseObject(o.toString());
            for (Map.Entry<String, Object> stringObjectEntry : jsonObject.entrySet()) {
                final Object entryValue = stringObjectEntry.getValue();
                final String filterType = stringObjectEntry.getKey();
                if(entryValue instanceof  JSONArray){
                    FilterList innerFilterList2;
                    if("or".equals(filterType)){
                        innerFilterList2=new FilterList(FilterList.Operator.MUST_PASS_ONE);
                    }else{
                        innerFilterList2=new FilterList(FilterList.Operator.MUST_PASS_ALL);
                    }
                    JSONArray filterA = JSONArray.parseArray(entryValue.toString());
                    filterA.forEach(o2->{
                        SingleColumnValueFilter singleColumnValueFilter = getSingleColumnValueFilter(o2);
                        innerFilterList2.addFilter(singleColumnValueFilter);
                    });
                    filterList.addFilter(innerFilterList2);
                }else{
                    SingleColumnValueFilter singleColumnValueFilter = getSingleColumnValueFilter(entryValue);
                    filterList.addFilter(singleColumnValueFilter);
                }
            }

        });


//        for (Map.Entry<String, Object> stringObjectEntry : mapscan.entrySet()) {
//            RowFilter rf = new RowFilter(CompareFilter.CompareOp.EQUAL,
//                    new BinaryComparator(Bytes.toBytes(stringObjectEntry.getValue().toString())));
//            filterList.addFilter(rf);
//        }
        filterList.setReversed(reversed);
        scan.setFilter(filterList);
        scan.setReversed(reversed);
        ResultScanner rs = table.getScanner(scan);
        //打印显示结果行的所有细胞
        List<Map<String, Object>> list = new ArrayList<>();
        getResult(rs, list);
        return list;
    }

    private   SingleColumnValueFilter getSingleColumnValueFilter(Object entryValue) {
        HbaseBaseFilter hbaseBaseFilter = JSONObject.parseObject(entryValue.toString(), HbaseBaseFilter.class);
        Object value = hbaseBaseFilter.getValue();
        final String filterType1 = hbaseBaseFilter.getFilterType();
        CompareFilter.CompareOp compareOp = getOpByFilterType(filterType1);
        SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter(Bytes.toBytes(hbaseBaseFilter.getFamily()), Bytes.toBytes(hbaseBaseFilter.getNoteKey()), compareOp, Bytes.toBytes(hbaseBaseFilter.getValue().toString()));
        //如果为 true，当这一列不存在时，不会返回，如果为 false，当这一列不存在时，会返回所有的列信息
        singleColumnValueFilter.setFilterIfMissing(true);
        return singleColumnValueFilter;
    }


    private    CompareFilter.CompareOp getOpByFilterType(String filterType) {
        switch (filterType) {
            case "equal":
                return CompareFilter.CompareOp.EQUAL;
            case "not_equal":
                return CompareFilter.CompareOp.NOT_EQUAL;
            case "lt":
                return CompareFilter.CompareOp.LESS;
            case "lte":
                return CompareFilter.CompareOp.LESS_OR_EQUAL;
            case "gt":
                return CompareFilter.CompareOp.GREATER;
            case "gte":
                return CompareFilter.CompareOp.GREATER_OR_EQUAL;
            default:
                return CompareFilter.CompareOp.NO_OP;

        }
    }
    private   void getResult(ResultScanner rs, List<Map<String, Object>> list) {
        for (Result r : rs) {
            List<Cell> cells = r.listCells();

            Map<String, Object> map = new HashMap<>();
            for (Cell cell : cells) {
                String row = Bytes.toString(CellUtil.cloneRow(cell));
                String cf = Bytes.toString(CellUtil.cloneFamily(cell));
                String cq = Bytes.toString(CellUtil.cloneQualifier(cell));
                String val = Bytes.toString(CellUtil.cloneValue(cell));

                map.put(cq, val);
            }
            list.add(map);
        }
    }


    public   boolean createTable(String tableName, String... families) {
        try {
            if(connection == null){
                getInstance();
            }
            Admin admin = connection.getAdmin();
            if (admin.isTableAvailable(TableName.valueOf(tableName))) {
                return true;
            }

            HTableDescriptor hBaseTable = new HTableDescriptor(TableName.valueOf(tableName));
            for (String family : families) {
                hBaseTable.addFamily(new HColumnDescriptor(family));
            }
            admin.createTable(hBaseTable);
            return true;
        } catch (IOException e) {
            log.error("", e);
            return false;
        }
    }

    public   String getStringVal(Result result, byte[] family, String col) {
        return Bytes.toString(CellUtil.cloneValue(result.getColumnLatestCell(family, Bytes.toBytes(col))));
    }

    /**
     * 判断表是否存在
     *
     * @param tableName 表名
     * @return bool
     */
    public   boolean isTableExist(String tableName) {
        try {
            if (connection == null) {
                getInstance();
            }
            Admin admin = connection.getAdmin();
            return admin.tableExists(TableName.valueOf(tableName));
        } catch (IOException e) {
            log.error("", e);
        }
        return false;
    }

    public   void close() {
        try {
            if (connection != null) {
                connection.close();
            }
        } catch (IOException e) {
            log.error("", e);
        }
    }
}
